package com.example.tcp.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.ResourceLeakDetector;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.Data;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.PreDestroy;
import java.util.function.Consumer;

import static io.netty.util.ResourceLeakDetector.Level.DISABLED;

/**
 * @Description: 接入服务器
 * @author: 林循益
 * @date: 2017/10/2 8:45
 * @Company: .
 * @version: V1.0
 */
@Data
@Slf4j
public class Server {

    /**
     * boss 线程组，负责accept
     */
    private EventLoopGroup bossGroup ;

    /**
     * work 线程组，负责处理clientIO
     */
    private EventLoopGroup workGroup ;

    /**
     * 解码线程，如果解码过程没有网络请求，可以将此线程数设置为0，直接在IO线程解难
     */
    private DefaultEventExecutorGroup decoderGroup;

    /**
     * 用户层业务逻辑线程组，主要处理用户业务逻辑，防止阻塞IO线程
     */
    private DefaultEventExecutorGroup bizGroup;

    /**
     * 服务器socket channel,负责accpet
     */
    private Channel serverChannel;

    /**
     * 启动服务器监听，并完成相关配置， 开始接受终端设备接入
     * @throws InterruptedException
     */
    public void start(Config config, Consumer<Channel> onInit) throws InterruptedException {
        this.bossGroup = createBossGroup();
        this.workGroup = createWorkGroup(config.getThreadCount());

        if(config.getDecoderThreadCount() > 0) {
            this.decoderGroup = new DefaultEventExecutorGroup(config.getDecoderThreadCount(), new DefaultThreadFactory("decode-group"));
        }

        this.bizGroup = new DefaultEventExecutorGroup(config.getBizThreadCount() > 0 ? config.getBizThreadCount() : 1, new DefaultThreadFactory("biz"));

        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workGroup).channel(getServerChannelClass());

        b.handler(new LoggingHandler(LogLevel.DEBUG))
                .option(ChannelOption.SO_BACKLOG, config.getBacklog())
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_LINGER, 0)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.SO_RCVBUF, config.getRecvBufSize())
                .childOption(ChannelOption.SO_SNDBUF, config.getSendBufSize())
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        if(config.getReadTimeoutSec() > 0) {
                            pipeline.addLast("ReadTimeoutHandler", new ReadTimeoutHandler(config.getReadTimeoutSec()));
                        }
                        onInit.accept(ch);
                    }
                });

        serverChannel = b.bind(config.getTcpPort()).sync().channel();

        // 内存泄漏检测
        ResourceLeakDetector.setLevel(config.getResDetectorLevel());

        log.info("tcp server start on port: {}", config.getTcpPort());
    }

    /**
     * bean在析构前先关闭服务器链路
     * @throws InterruptedException
     */
    @PreDestroy
    public void stop() throws InterruptedException {
        if(serverChannel != null) {
            serverChannel.close().sync();
        }
        bossGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }

    /**
     * 创建boss线程组,默认为1个线程
     * @return 线程组
     */
    public static EventLoopGroup createBossGroup() {
        return createEventLoopGroup(100, 0);
    }

    /**
     * 创建工作线程组
     * @param nThreads 工作线程数
     * @return 线程组
     */
    public static EventLoopGroup createWorkGroup(int nThreads) {
        return createEventLoopGroup(80, nThreads);
    }

    /**
     * 创建线程组,根据平台自行判断使用epoll,nio
     * @param ioRatio io执行率
     * @param nThreads 线程数
     * @return 线程组 EventLoopGroup
     */
    public static EventLoopGroup createEventLoopGroup(int ioRatio, int nThreads) {
        if (Epoll.isAvailable()) {
            EpollEventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup(nThreads);
            epollEventLoopGroup.setIoRatio(ioRatio);
            return epollEventLoopGroup;
        }
        NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(nThreads);
        nioEventLoopGroup.setIoRatio(ioRatio);
        return nioEventLoopGroup;
    }


    /**
     * 获取服务类工厂
     * @return ServerSocketChannel
     */
    public static Class<? extends ServerSocketChannel> getServerChannelClass() {
        if (Epoll.isAvailable()) {
            return EpollServerSocketChannel.class;
        }
        return NioServerSocketChannel.class;
    }

    /**
     * @Description: netty服务配置类
     * @author: 林循益
     * @date: 2017/10/2 9:12
     * @Company: .
     * @version: V1.0
     */
    @Data
    @ToString
    @Accessors(chain = true)
    public static class Config {
        /**
         * netty work 线程数，为0则由netty自动计算，一般是系统核心数*2
         */
        private int threadCount = 0;

        /**
         * netty 监听端口
         */
        private int tcpPort = 9021;

        /**
         * client socket接收buffer大小
         */
        private int recvBufSize = 1024;

        /**
         * client socket发送buffer大小
         */
        private int sendBufSize = 1024;

        /**
         * socket读超时时间,单位秒
         */
        private int readTimeoutSec = 180;

        /**
         * 解码线程数，注意如果解码不需要进行网络请求，请将此值设置为0
         */
        private int decoderThreadCount = 0;

        /**
         * biz work 线程数,一般不要设置为0，以防业务线程阻塞netty IO
         */
        private int bizThreadCount = 1;

        /**
         * server socket accept 缓冲队列
         */
        private int backlog = 1024;

        /**
         * netty 内在泄露检测等级，生产环境请禁用
         */
        private ResourceLeakDetector.Level resDetectorLevel = DISABLED;
    }
}
